package org.shj.spark.core;

import java.util.Arrays;
import java.util.Iterator;
import java.util.List;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.PairFunction;

import scala.Tuple2;

public class TopNGroup {

	public static void main(String[] args) {
		SparkConf conf = new SparkConf();
		conf.setAppName("TopNGroup").setMaster("local");
		
		JavaSparkContext sc = new JavaSparkContext(conf);
		JavaRDD<String> lines = sc.textFile("E:/workspace/scala/sparkjava/src/main/resources/topNGrp.txt");
		
		JavaPairRDD<String, Tuple2<Integer,String>> pair = lines.mapToPair(new PairFunction<String, String, Tuple2<Integer,String>>() {

			private static final long serialVersionUID = 7245523132397418925L;

			@Override
			public Tuple2<String, Tuple2<Integer, String>> call(String line) throws Exception {
				String[] arr = line.split("\\s+");
				Tuple2<Integer, String> val = new Tuple2<Integer, String>(Integer.parseInt(arr[2]), line);
				return new Tuple2<String, Tuple2<Integer, String>>(arr[0], val);
			}	
			
		});
		
		JavaPairRDD<String, Iterable<Tuple2<Integer,String>>> grpPairs = pair.groupByKey();
		JavaPairRDD<String, Iterable<Tuple2<Integer, String>>> topN = grpPairs.mapToPair(new PairFunction<Tuple2<String,Iterable<Tuple2<Integer,String>>>, String, Iterable<Tuple2<Integer,String>>>() {
			private static final long serialVersionUID = 6768364468361188382L;

			public Tuple2<String, Iterable<Tuple2<Integer,String>>> call(Tuple2<String, Iterable<Tuple2<Integer,String>>> grpPair) throws Exception {
				int topN = 2;
				Tuple2<Integer,String>[] topNArr = new Tuple2[2]; //假设只需要top 2
				Iterator<Tuple2<Integer,String>> grpVals = grpPair._2.iterator();
				
				while(grpVals.hasNext()) {
					Tuple2<Integer, String> val = grpVals.next();
					for(int i = 0 ; i < topN; i++) {
						if(topNArr[i] == null) {
							topNArr[i] = val;
							break;
						}else if(val._1 > topNArr[i]._1){
							for(int j = topN - 1; j > i; j--) {
								topNArr[j] = topNArr[j-1];
							}
							topNArr[i] = val;
							break;
						}
					}
				}
				
				return new Tuple2<String, Iterable<Tuple2<Integer,String>>>(grpPair._1, Arrays.asList(topNArr));
			}
			
		});
		
		List<Tuple2<String, Iterable<Tuple2<Integer, String>>>> collect = topN.collect();
		for(Tuple2<String, Iterable<Tuple2<Integer, String>>> kv : collect) {
			System.out.println("GroupedBy: " + kv._1);
			Iterator<Tuple2<Integer, String>> v = kv._2.iterator();
			while(v.hasNext()) {
				System.out.println(v.next()._2);
			}
		}
		
		sc.close();
		
	}

}
